
package com.lee.job; 
 
import com.alibaba.fastjson.JSON; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.ConstantsUtils; 
import com.lee.utils.DMapUtils; 
import org.apache.flink.api.common.functions.RichMapFunction; 
import org.apache.flink.api.common.state.MapState; 
import org.apache.flink.api.common.state.MapStateDescriptor; 
import org.apache.flink.configuration.Configuration; 
 
public class StoreDMapFunction extends RichMapFunction<CkFlowInfo, CkFlowInfo> { 
    private MapState<String,Long> dMapState;
    @Override
    public void open(Configuration parameters){
        dMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String,Long>("dmap",String.class,Long.class));
    }
 
 
    @Override
    public CkFlowInfo map(CkFlowInfo ckFlowInfo) throws Exception {
        try {
 
            Logger.debug(ckFlowInfo.getDIndex(),"/",ckFlowInfo.getDCIndex(),"StoreDmapFunction ->  rawCkFlowInfo ", ckFlowInfo);
 
 
            long index = ckFlowInfo.getDIndex();
 
 
            ckFlowInfo.getCkList().sort(String::compareTo);
            String lockKey = JSON.toJSONString(ckFlowInfo.getCkList());
            long lockValue = index;
 
            //判断是否存在锁
            long waitTimes = 60;
            while(waitTimes > 0 && dMapState.contains(lockKey) && lockValue != dMapState.get(lockKey)){
 
                waitTimes -= 1;
                Logger.info(ckFlowInfo.getDIndex(),"/",ckFlowInfo.getDCIndex(),"StoreDmapFunction ->  存在锁，剩余",waitTimes,"次尝试，等待一段时间后重新获取 ", ckFlowInfo.getCkList(),index);
                Thread.sleep(1000);
            }
            //加锁
            dMapState.put(lockKey,lockValue);
 
            DMapUtils.addItemToMapGinix(ckFlowInfo.getCkList(), index);
 
            //释放锁
            dMapState.remove(lockKey);
 
 
            DMapUtils.addFinishIndex(FinishQualifierEnum.STORE_DMAP.getCode(),1l);
            Logger.debug(ckFlowInfo.getDIndex(),"/",ckFlowInfo.getDCIndex(),"StoreDmapFunction ->  finished  add ckFlowInfo ", ckFlowInfo.getCkList(),index);
 
 
 
        }catch (Exception e){
            //释放锁
            String lockKey = JSON.toJSONString(ckFlowInfo.getCkList());
            if(dMapState.contains(lockKey)){
                dMapState.remove(lockKey);
            }
            Logger.error("StoreDmapFunction -> 初始化数据报错：",e);
            e.printStackTrace();
        }
        return ckFlowInfo;
    }
} 
